package transformationPractice;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;


import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

@SuppressWarnings(value={"unused","unchecked"})
public class TransformationOperation {
    public static JavaSparkContext  sc = null;
    public static void main(String[] args) {
//        map();
//        filter();
//        flatMap();
//        groupByKey();

        reduceByKey();
    }

    public static JavaSparkContext init(String name){
        // 创建 SparkConf
        SparkConf conf = new SparkConf()
                .setAppName(name)
                .setMaster("local");

        return new JavaSparkContext(conf);

    }

    /**
     * map 算子案例： 将集合中的一个元素乘以 2
     */
    private static void map(){
        // 创建 sparkContext
        sc = init("map");
        // 构造集合
        List<Integer> number = Arrays.asList(1,2,3,4,5);

        // 并行化操作集合， 创建初始RDD
        JavaRDD<Integer> numberRDD = sc.parallelize(number);

        // 使用map 算子， 将集合中的每个元素都乘以 2
        // map 算子， 是对任何类型的RDD 都可以调用的
        // 在Java中， map 算子接受的参数是 Function 对象
        // 之后 创建的 Function对象， 一定会让你设置第二个泛型参数， 这个泛型类型， 就是返回的新元素的类型
        // 同时 call 方法的返回类型，也必须与第二个泛型类型同步一样
        // 在 call 方法内部， 就可以对原始的 RDD 中的每一个元素进行各种处理和计算， 并返回一个新的元素
        // 所有的元素就会组成一个新的 RDD
        JavaRDD<Integer> multipleNumberRDD = numberRDD.map(new Function<Integer, Integer>() {
            @Override
            public Integer call(Integer v1) throws Exception {
                return v1 * 2;
            }
        });

        multipleNumberRDD.foreach(new VoidFunction<Integer>() {
            @Override
            public void call(Integer value) throws Exception {
                // 打印
                System.out.println(value);
            }
        });
        // 关闭 JavaContext
        sc.close();
    }

    /**
     * filter算子案例：过滤集合中的偶数
     */
    private static void filter(){
        sc = init("filter");

        // 模拟集合
        List<Integer> numbers = Arrays.asList(1,2,3,4,5,6,7,8,9);

        // 并行化集合创建初始RDD
        JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

        // 对初始化的 RDD 执行filter 算子
        /* filter算子，传入的也是 Function, 其他的使用注意点，实际上和map一样
            但是 ， 唯一不同的是 call() 方法 返回的类型是 Boolean
            每一个处处RDD中的数据元素， 都会传入call()方法， 此时你就可以根据各种自定义ID计算逻辑
            来判断这个元素是否是你想要的
            如果是你想在新的RDD中保留这个元素， 那么久返回true， 否则，不想保留这个应荣生 返回false
         */

        JavaRDD<Integer> evenNumber = numberRDD.filter(new Function<Integer, Boolean>() {
            @Override
            public Boolean call(Integer value) throws Exception {
                return value % 2 == 0;  // 保留 偶数 此时情况 为 true 的情况 会返回偶数
            }
        });

        // 打印
        evenNumber.foreach(new VoidFunction<Integer>() {
            @Override
            public void call(Integer value) throws Exception {
                System.out.println("保留的偶数： "+ value);
            }
        });
        // 关闭
        sc.close();
    }

    /**
     * flatMap 案例： 将文本行 拆分成多个单词
     */
    private static void flatMap(){
        sc = init("flatMap");

        // 构建集合
        List<String> list = Arrays.asList("hello you", "hello me", "hello world");

        // 对 RDD 执行 Transformation 算子操作， 将每每一行 文本， 拆分分成多个单词
        JavaRDD<String> lines = sc.parallelize(list);
        
        // 并行化处理 RDD 执行faltMap算子， 将每一行文本， 拆分为多个单词
        /*
        flatMap算子 ， 在Java中， 接收到参数是FlatMapFunction
        我们需要自己定义 FlatMapFunction 的第二个泛型参数， 即 代表了返回的新元素的类型
        call()方法，返回的类型， 不是 U, 而是Iterable<U>, 这里的 U 也与第二个泛型类型相同
        flatMap 其实就是： 接受原始RDD中的每个元素， 并进行各种逻辑的计算和处理， 可以返回的多个元素
        这多个元素，即是 封装在 Iterable集合中， 可以使用 ArrayList 等集合中

        新的RDD中，一定封装了 所有的新元素， 也就是说，新的RDD的大小一定 >= 原始RDD的大小
        */
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            // 在这里比如传入 第一个 RDD "hello you", 结果就会是 hello 和 you
            @Override
            public Iterator<String> call(String s) throws Exception {
                List<String> strings = Arrays.asList(s.split(" "));
                return strings.iterator();
            }

            /**
             * public Iterable<String> call(String s) throws Exception {
             *                 return Arrays.asList(s.split(" "));;
             *             }
             */
        });

        words.foreach(new VoidFunction<String>() {
            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });

        // 关闭 JavaSparkContext
        sc.close();
    }

    /**
     * groupByKey 实际操作 : 将每个班级的成绩进行分组
     */

    private static void groupByKey(){
        // 创建 sparkconf 和 JavaSparkContext
        sc = init("groupByKey");

        // 模拟集合  其中 Tuple 中第一个元素表示 班级 第二个表示 对应的 成绩
        List<Tuple2<String, Integer>> scoreList = Arrays.asList(
                new Tuple2<>("class1", 80),
                new Tuple2<>("class2", 90),
                new Tuple2<>("class1", 72),
                new Tuple2<>("class2", 77),
                new Tuple2<>("class1", 93),
                new Tuple2<>("class2", 124)
                );

        // 并行化集合  下面这样写 无法直接操作 groupByKey
        // 为什么这样做， 这是因为 scoreList 并不是 RDD 类型的数据， 需要处理成 RDD类型数据， 使用 parpallelize可以进行处理
        JavaRDD<Tuple2<String, Integer>> scores = sc.parallelize(scoreList);
        // 或者下面这样写
        JavaPairRDD<String, Integer> scores2 =  sc.parallelizePairs(scoreList);
        // groupByKey 是用在 key-value 类型上面的
        /**
         * groupByKey 算子， 返回的还是 JavaRDD,
         * 但是， JavaPairRDD的第一个泛型参数不变， 第二泛型参数编程 Iterable 集合类型
         * 也就是说， 按照 key 进行分组，那么 每个 key 可能都会有多个 value, 此时
         * 多个value 聚合成立 Iterable， 那么接下来， 我们是不是就可以通过 groupedScores 这种
         * JavaRDD 方便地处理某个分组内的数据
         */

        JavaPairRDD<String, Iterable<Integer>> groupedScores = scores2.groupByKey();

        // 打印新的groupScores RDD
        groupedScores.foreach(new VoidFunction<Tuple2<String, Iterable<Integer>>>() {
            @Override
            public void call(Tuple2<String, Iterable<Integer>> t) throws Exception {
                System.out.println("class: "+t._1);
                Iterator<Integer> it = t._2.iterator();
                while (it.hasNext()){
                    System.out.println(it.next());
                }

                System.out.println("=====================");
            }
        });


        // 关闭JavaSparkContext
        sc.close();
    }

    /** 5
     * reduceByKey 实际操作 : 统计每个班级的总分
     */

    private static void reduceByKey(){
        // 创建 sparkconf 和 JavaSparkContext
        sc = init("groupByKey");

        // 模拟集合  其中 Tuple 中第一个元素表示 班级 第二个表示 对应的 成绩
        List<Tuple2<String, Integer>> scoreList = Arrays.asList(
                new Tuple2<>("class1", 80),
                new Tuple2<>("class2", 90),
                new Tuple2<>("class1", 72),
                new Tuple2<>("class2", 77),
                new Tuple2<>("class1", 93),
                new Tuple2<>("class2", 124)
        );

        // 并行化集合
        JavaPairRDD<String, Integer> Scores =  sc.parallelizePairs(scoreList);
        // groupByKey 是用在 key-value 类型上面的

        /**
         * 针对 scores RDD, 执行reduceByKey 算子
         * reduceByKey 接收的参数是 Function2 类型， 他有三个参数， 实际上代表了三个值
         * 第一个泛型参数 和 第二个泛型参数类型 代表了原始 RDD 中的元素的 value 类型
         * 因为 对每个 key 进行 reduce, 都会依次将 第一个、第二个 value 传入，
         * 将此 值 再与 第三个 value 传入，因此此处， 会自动定义两个泛型类型， 代表call()方法的
         * 两个传入参数的类型
         * 第三个泛型类型， 代表了每次 reduce 操作返回的值的类型， 默认也是与原始RDD的value值类型相同的
         * reduceByKey算法返回的RDD，还是 JavaPairRDD<key, value>
         */

        JavaPairRDD<String, Integer> totalScores = Scores.reduceByKey(
                new Function2<Integer, Integer, Integer>() {
                    @Override
                    public Integer call(Integer v1, Integer v2) throws Exception {
                        return v1 + v2;
                    }
                }
        );

        totalScores.foreach(
                new VoidFunction<Tuple2<String, Integer>>() {
                    @Override
                    public void call(Tuple2<String, Integer> t) throws Exception {
                        System.out.println(t._1+" : "+t._2);
                    }
                }
        );

        // 关闭 SparkContext
        sc.close();
    }

    /**
     * sortByKey: 将学生分数进行排序
     */

    private static void sortByKey(){
        // 创建 sparkconf 和 JavaSparkContext
        sc = init("groupByKey");

        // 模拟集合  其中 Tuple 中第一个元素表示 班级 第二个表示 对应的 成绩
        List<Tuple2<Integer, String>> scoreList = Arrays.asList(
                new Tuple2<>(65,"leo"),
                new Tuple2<>(50,"Tom"),
                new Tuple2<>(100,"mary"),
                new Tuple2<>(60,"jack")

        );

        // 并行化集合
        JavaPairRDD<Integer, String> scores =  sc.parallelizePairs(scoreList);

        // 对  scores RDD 执行 sortByKey算子
        JavaPairRDD<Integer, String> sortedScores = scores.sortByKey();

        // 打印 sortedScore
        sortedScores.foreach(
                new VoidFunction<Tuple2<Integer, String>>() {
                    @Override
                    public void call(Tuple2<Integer, String> t) throws Exception {
                        System.out.println("分数："+t._1+"  : 学生 "+ t._2);
                    }
                }
        );
        sc.close();

    }

}
